package org.nbict.iot.task.once.schedule;

import com.alibaba.fastjson.JSON;
import org.nbict.iot.Pair;
import org.nbict.iot.persist.pool.DbRepository;
import org.nbict.iot.persist.repo.RedisRepo;
import org.nbict.iot.task.once.Pointer;
import org.nbict.iot.task.once.TimeSpanWindow;

import java.util.Date;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * Created by songseven on 18/7/12.
 */
public class RunningScheduler implements Runnable {

    private RedisRepo repo;
//    private MongoRepo mongoRepo;
//    private MysqlRepo mysqlRepo;
    private int page;

    private Long delay; //3天

    private Long count;

    public RunningScheduler() {
        repo = (RedisRepo) DbRepository.REDIS.getInstance();
        page = 1000;
        this.delay = 3*24*3600l; //3天
//        mongoRepo = (MongoRepo) DbRepository.MONGO.getInstance();
//        mysqlRepo = (MysqlRepo) DbRepository.MYSQL.getInstance();

        count = 0l;
    }

    @Override
    public void run() {
        compress();
    }

    //压缩
    private void compress() {
        System.out.println(new Date() +  " => begin compress...");
        Set<String> cars = repo.getKeys("[a-zA-Z_/d]*"); //默认为知豆 LJU70X1Z8HG552089 LJU70X1Z9HG103133

        //防止内存溢出，分页
        cars.forEach(car->{

            System.out.println(car);

            int pageCount = 0;
            Long size = repo.getSize(car);
            if ((pageCount = (int)(size / page)) > 0 && size % page > 0) {
                pageCount++;
            } else if (pageCount == 0){
                pageCount = 1;
            }

            TimeSpanWindow journey = new TimeSpanWindow(car, repo); //计算窗口
            for (int i = 0, rem = 0, total = 0; pageCount > 0; pageCount--, i++) {
                if (pageCount == 1) {
                    rem = (int)(size % page);
                } else {
                    rem = page;
                }

                Set<String> pointers = repo.getZSet(car, (long)i * page, (long)(i * page + rem - 1));
                if (pointers != null) {
                    for(Iterator it = pointers.iterator(); it.hasNext(); total++){
                        String tuple = (String)it.next();
                        Pointer p = JSON.parseObject(tuple, Pointer.class);
                        //计算点
                        journey.take(p);

                        if (total == size - 1 && total > 0) { //最后一个元素
                            Pair<Pointer, Pointer> remind = journey.getRemind();
                            remind.setRight(p);
                            journey.getRemindHolder().add(
                                    new Pair<Long, Long>(remind.getLeft().currentTime(),
                                            remind.getRight().currentTime()));
                            remind.clear();
                        }
                    }
                }
            }

            //System.out.println("compressing...");
            //压缩行程
            journey.getRemindHolder().forEach(remind -> {
                repo.removeWithin(car, remind.getLeft(), remind.getRight());
            });

//            //System.out.println("compute ==> charged");
//            //计算充电
//            List<TimeSpan> times = new ArrayList<>();
//            journey.chargeHolder().forEach(span -> {
//                //debug
//                Long current = System.currentTimeMillis() / 1000;
//                if (current - span.getRight()
//                        .currentTime() > delay) {
//                    TimeSpan time = new TimeSpan(car,
//                            span.getLeft().currentTime(), //起始时间 有可能为null
//                            span.getRight().currentTime(),//结束时间
//                            span.getLeft().getLon(),//起始经度
//                            span.getLeft().getLat(),//起始纬度
//                            span.getRight().getLon(),//结束经度
//                            span.getRight().getLat(),//结束纬度
//                            span.getRight().getSoc() - span.getLeft().getSoc(), //soc变化
//                            "C"); //充电状态
//                    times.add(time);
//                }
//            });
//
//            //System.out.println("insert ==> charged");
//            //插入充电记录
//            if (!times.isEmpty()) {
//                mysqlRepo.insertJourney(times);
//            }
        });

        System.out.println("end compress...");
    }

//    public void removeWithinTimestamp(String car, Long from, Long to) {
//        repo.removeWithinTimestamp(car, from, to);
//    }

    public static void main(String[] args) throws InterruptedException {

        ExecutorService executorService = Executors.newScheduledThreadPool(1);
        for (int i = 0; i < 1; i++) {
            executorService.submit(new RunningScheduler());
        }

        Runtime.getRuntime().addShutdownHook(new Thread() {
            @Override
            public void run() {
                //TODO
                executorService.shutdown();
            }
        });
    }
}
